package com.heimachengxuyuan.base.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 消息的接受者
 * 这里看消费的结果  好像不是顺序消费
 *
 * 要测试负载均衡或者是广播的效果，那么可以启动多个consumer，起多个实例
 */
public class Consumer {

    public static void main(String[] args) throws Exception {
        //1.创建消费者Consumer，制定消费者组名
        //对哦，这里为什么是 PushConsumer 呢？？？这是broker给消费者推的模式
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");
        //3.订阅主题Topic和Tag（这个subExpression就是tag，这是一个表达式，也可以写成*）
        consumer.subscribe("base", "Tag1 || Tag2");
        //设定消费模式：要么就是负载均衡，要么就是广播模式、不设置的话，默认是负载均衡模式
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.设置消息监听器，也就是回调
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    //消费的时候是用多个线程去消费
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
    }
}
